package com.akkainaction.cluster.wordcount.router;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.dispatch.OnFailure;
import akka.dispatch.OnSuccess;
import akka.pattern.Patterns;
import akka.routing.FromConfig;
import akka.util.Timeout;
import com.akkainaction.cluster.wordcount.router.po.Article;
import com.akkainaction.cluster.wordcount.router.po.CountResult;
import com.akkainaction.cluster.wordcount.router.service.WordCountService;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class StartClient {

    public static void main(String[] args) {
        String directory = "D:\\mycode\\akka-in-action-demo\\data";
        Config config = ConfigFactory.load("wordcount-router-front.conf");
        ActorSystem system = ActorSystem.create("sys", config);
        ActorRef router = system.actorOf(FromConfig.getInstance().props(Props.create(WordCountService.class)), "wordCountServiceActor");
        List<Article> arts = new ArrayList<>();
        File dir = new File(directory);
        File[] files = dir.listFiles();
        try {
            for (File file : files) {
                Article art = new Article();
                art.setId(file.getName());
                StringBuffer contentBf = new StringBuffer();
                BufferedReader reader = new BufferedReader(new FileReader(file));
                String line = reader.readLine();
                while (line != null) {
                    contentBf.append(line);
                    line = reader.readLine();
                }
                reader.close();
                art.setContent(contentBf.toString());
                arts.add(art);
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }
        Timeout timeout = new Timeout(Duration.create(3, TimeUnit.SECONDS));

        try {
            //睡眠用来等待集群建立完整
            TimeUnit.SECONDS.sleep(30);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        for (Article art : arts) {
            System.out.println(args);
            Future<Object> ask = Patterns.ask(router, art, timeout);
            ask.onSuccess(new OnSuccess<Object>() {
                @Override
                public void onSuccess(Object res) throws Throwable {
                    CountResult cr = (CountResult) res;
                    System.out.println("文件" + cr.getId() + ",单词数:" + cr.getCount());
                }
            }, system.dispatcher());
            ask.onFailure(new OnFailure() {
                @Override
                public void onFailure(Throwable failure) {
                    System.out.println(failure.getCause());
                }
            },system.dispatcher());
        }
    }
}
